import json
import os
import threading
import time
import queue
from kafka import KafkaConsumer, KafkaProducer
from util import format_conversion
from process_event import ProcessEvent
from process_congestion import ProcessCongestion
from datetime import datetime, timedelta


class Event:
    def __init__(self, configData, define, kakoList, directory):
        self.define = define
        self.config_data = configData
        self.format_str = self.config_data["format_str"]
        self.gantry_info = self.config_data["gantry_info"]
        self.route_list = self.config_data["route"]
        self.save_output_path = os.path.join(directory, "log", "output")
        self.save_json_path = os.path.join(directory, "log", "json")
        self.config_data["save_output_path"] = self.save_output_path
        self.config_data["save_json_path"] = self.save_json_path
        # 存储门架状态信息
        self.gantry_status = {}
        # 存储门架所有数据
        self.gantry_data = {}
        # 多门架数据队列
        self.q = queue.Queue(maxsize=1000)
        # 拥堵数据队列
        self.q_congestion = queue.Queue(maxsize=1000)
        # 单区间门架数据队列
        self.q_dict = {}
        # 线程运行标志队列，用于判断是否结束线程
        self.flag_dict = {}
        # 阻断信息上报存储队列
        self.block_dict = {}
        # 拥堵信息上报存储队列
        self.congestion_queue = queue.Queue(maxsize=1000)
        # 拥堵卡口数据队列
        self.congestion_kako_queue = queue.Queue(maxsize=1000)
        # 拥堵雷达数据队列
        self.congestion_radar_queue = queue.Queue(maxsize=1000)
        # 门架监测数据上报存储队列
        self.kafka_data_dict = {}
        # 门架编号字典，每个键为门架编号，值为门架对应的区间编号
        self.id_dict = {}
        # 存储门架经纬度等信息
        self.kako_dict = {item['pileNumber']: {key: value for key, value in item.items() if key != 'pileNumber'} for item in kakoList}
        # # 各区间上报时的事件ID（eventId）字典
        # self.event_id_dict = {"TLS_1_1": 2001, "TLS_1_2": 2002, "TLS_2_1": 2003, "TLS_2_2": 2004, "ZW4": 2005,
        #                       "ZW3": 2006, "HW3": 2007, "TLS_3_1": 2008, "TLS_3_2": 2009, "TLS_4_1": 2010,
        #                       "TLS_4_2": 2011, "ZW1": 2012, "ZW2": 2013, "HW6": 2014}
        # self.congestion_id_dict = {"TLS_1_1": 1001, "TLS_1_2": 1002, "TLS_2_1": 1003, "TLS_2_2": 1004, "ZW4": 1005,
        #                            "ZW3": 1006, "HW3": 1007, "TLS_3_1": 1008, "TLS_3_2": 1009, "TLS_4_1": 1010,
        #                            "TLS_4_2": 1011, "ZW1": 1012, "ZW2": 1013, "HW6": 1014}
        # 使用字典存储每个区间的事件计数器  结构: {road_section: {"congestion": counter, "blockage": counter}}
        self.counters = {}
        # 事件类型映射
        self.event_types = {
            "congestion": 1,  # 拥堵类型代码
            "block": 2  # 阻断类型代码
        }
        # 存储当前活跃事件 {road_section: event_data}
        self.active_events = {}
        # 线程管理
        self.thread_flag = False
        # 日志保存地址
        self.folder_path = None
        self.folder_json_path = None
        # 保存开始时间
        self.start_time = None
        # kafka 服务器
        self.kafka_producer = KafkaProducer(
            bootstrap_servers=self.config_data["kafka_host"],
            key_serializer=str.encode,
            value_serializer=lambda x: json.dumps(x).encode('utf-8')
        )
        self.kafka_consumer_congestion = KafkaConsumer(
            self.config_data["kafka_event_topic"],
            bootstrap_servers=self.config_data["kafka_host"],
            auto_offset_reset='latest',
            enable_auto_commit=False
        )

    def subscribe_data_kafka(self):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        print("门架数据监听启动")
        kafka_consumer = KafkaConsumer(
            self.config_data["kafka_cars_topic"],
            bootstrap_servers=self.config_data["kafka_host"],
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        while True:
            time.sleep(0.02)
            for message in kafka_consumer:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        vehicle_info = json.loads(message_value)
                        self.q.put(vehicle_info, block=False)
                    except Exception as e:
                        print("解析数据失败", datetime.now(), "topic:", self.config_data["kafka_cars_topic"], e)
            # 退出线程
            if self.thread_flag:
                break

    def subscribe_device_data_kafka(self):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        print("门架状态数据监听启动")
        kafka_device_congestion = KafkaConsumer(
            self.config_data["kafka_all_status"],
            bootstrap_servers=self.config_data["kafka_host"],
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        while True:
            time.sleep(0.02)
            for message in kafka_device_congestion:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        data = json.loads(message_value)
                        data_info = data["gantryInfos"]
                        for i in range(len(data_info)):
                            gantry = data_info[i]["sn"]
                            heartbeat = data_info[i]["heartbeat"]
                            clock = data_info[i]["clock"]
                            delay_status = data_info[i]["delay_status"]
                            precision_status = data_info[i]["precision_status"]
                            self.gantry_status[gantry] = [heartbeat, clock, delay_status, precision_status]
                    except Exception as e:
                        print("解析数据失败", datetime.now(), "topic:", self.config_data["kafka_all_status"], e)
            # 退出线程
            if self.thread_flag:
                break

    # def subscribe_congestion_data_kafka(self):
    #     # 将消费者的偏移量设置为最后一条消息的偏移量
    #     print("拥堵上报信息监听启动")
    #     while True:
    #         time.sleep(0.02)
    #         for message in self.kafka_consumer_congestion:
    #             if message:
    #                 try:
    #                     message_value = message.value.decode('utf-8')
    #                     congestion_info = json.loads(message_value)
    #                     self.q_congestion.put(congestion_info, block=False)
    #                 except Exception as e:
    #                     print("解析状态数据失败", datetime.now(), "topic:", self.config_data["kafka_event_topic"], e)
    #         # 退出线程
    #         if self.thread_flag:
    #             break

    def subscribe_radar_data(self, topic, flag):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        print("radar上报信息监听启动")
        kafka_consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.config_data["kafka_host"],
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        while True:
            time.sleep(0.02)
            for message in kafka_consumer:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        congestion_info = json.loads(message_value)
                        if flag == "radar":
                            congestion_info["ip"] = topic.split("_")[1]
                        self.congestion_radar_queue.put(congestion_info, block=False)
                    except Exception as e:
                        print("解析雷达数据失败", datetime.now(), "flag:", flag, "topic:", topic, e)
            # 退出线程
            if self.thread_flag:
                break

    def init(self):
        # 存储区间编号和对应的门架编号的字典 {区间编号: [上游门架编号,下游门架编号]}
        info_dict = {}
        # 创建当前时间的文件夹
        T = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        self.folder_path = os.path.join(self.save_output_path, T)
        if not os.path.exists(self.folder_path):
            os.makedirs(self.folder_path)
        self.folder_json_path = os.path.join(self.save_json_path, T)
        if not os.path.exists(self.folder_json_path):
            os.makedirs(self.folder_json_path)

        self.start_time = datetime.now()
        # 根据区间编号初始化
        for info in self.gantry_info.keys():
            self.q_dict[info] = queue.Queue(maxsize=1000)
            self.flag_dict[info] = queue.Queue(maxsize=10)
            self.block_dict[info] = queue.Queue(maxsize=1000)
            self.kafka_data_dict[info] = queue.Queue(maxsize=1000)

            info_dict[info] = [self.gantry_info[info]["up_gantry_id"], self.gantry_info[info]["down_gantry_id"]]
        # 获取门架ID对应的区间编号
        self.id_dict = format_conversion(info_dict)
        for gantry in self.id_dict.keys():
            self.gantry_status[gantry] = [1, 1, 1, 1]
        print("初始化完成！！！")
        # 实时监听最新门架数据
        threading.Thread(target=self.subscribe_data_kafka).start()
        # 实时监听最新拥堵上报信息
        # threading.Thread(target=self.subscribe_congestion_data_kafka).start()
        # 实时监听最新门架状态数据
        threading.Thread(target=self.subscribe_device_data_kafka).start()
        # 实时监听雷达数据
        if self.config_data["radar_flag"]:
            for radar_topic in self.config_data["radar_topic"]:
                threading.Thread(target=self.subscribe_radar_data, args=(radar_topic, "radar")).start()
            for lidar_topic in self.config_data["lidar_topic"]:
                threading.Thread(target=self.subscribe_radar_data, args=(lidar_topic, "lidar")).start()

        # 创建多个线程事件，每个线程处理一个区间
        for info in self.gantry_info.keys():
            config_data = self.config_data
            config_data["gantry_info"] = self.gantry_info[info]
            config_data["gantry_info"]["info"] = info
            config_data["gantry_info"]["time_interval"] = self.config_data["all_time_interval"]
            p = threading.Thread(target=self.process_main, args=(config_data, info,))
            p.start()

        # 启动拥堵处理线程
        threading.Thread(target=self.process_congestion).start()

        print("多线程启动完成")

    def generate_id(self, road_section_str, event_type, event_time, event_source):
        # 确保road_section是整数
        road_section = int(road_section_str)
        # 验证事件类型
        if event_type not in self.event_types:
            raise ValueError(f"无效的事件类型: {event_type}。请使用'congestion'或'block'")
        # 初始化计数器(如果不存在)
        if road_section not in self.counters:
            self.counters[road_section] = {"congestion": 0, "block": 0}
            self.active_events[road_section] = {}
        # 获取当前计数器值并递增
        if (self.counters[road_section][event_type] > 0) and (event_type in self.active_events[road_section].keys()):
            last_event_source = self.active_events[road_section][event_type]["event_source"]
            event_duration = event_time - self.active_events[road_section][event_type]["event_time"]
            min_duration = timedelta(minutes=float(self.gantry_info[road_section_str]['time_move']) * 1.5)
            if last_event_source != event_source or event_duration > min_duration:
                self.counters[road_section][event_type] += 1
                self.active_events[road_section][event_type]["start_time"] = event_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
        else:
            self.counters[road_section][event_type] += 1
            self.active_events[road_section][event_type] = {}
            self.active_events[road_section][event_type]["start_time"] = event_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
        # 检查是否超出范围(确保不超过99)
        if self.counters[road_section][event_type] > 99:
            self.counters[road_section][event_type] = self.counters[road_section][event_type] // 99
        current_counter = self.counters[road_section][event_type]
        # 生成ID: 道路编号×1000 + 类型代码×100 + 序列号
        event_id = (road_section * 1000) + (self.event_types[event_type] * 100) + current_counter
        # 检查是否超过int最大值(虽然Python int理论上无限大，但为了兼容其他系统)
        if event_id > 2147483647:
            event_id = event_id // 2147483647
            print("event_id超出int最大值，将使用除法取余后的值")

        self.active_events[road_section][event_type]["event_id"] = event_id
        self.active_events[road_section][event_type]["event_time"] = event_time
        self.active_events[road_section][event_type]["event_source"] = event_source

        return event_id

    def process_main(self, config_data, info):
        # 创建ProcessEvent实例，传入配置数据进行初始化
        process_event = ProcessEvent(config_data, self.define)
        # 初始化进程事件，进行必要的设置和准备
        process_event.init()
        # 启动进程事件，开始处理任务
        process_event.run(self.q_dict[info], self.block_dict[info], self.kafka_data_dict[info], self.flag_dict[info])

    def process_congestion(self):
        first_info = list(self.gantry_info.keys())[0]
        # 创建ProcessCongestion实例，传入配置数据进行初始化
        process_congestion = ProcessCongestion()
        # 启动线程事件，开始处理任务
        process_congestion.run(self.congestion_radar_queue, self.congestion_kako_queue, self.congestion_queue,  self.flag_dict[first_info])

    def add_message(self):
        sn_info_dict = {}
        for info in self.gantry_info.keys():
            sn_info_dict[self.gantry_info[info]["down_gantry_id"]] = info
        level_dict = {"畅通": 0, "轻度拥堵": 1, "中度拥堵": 2, "严重拥堵": 3, "阻断": 4, "拥堵+阻断": 5}
        # 创建当前时间的文件，用于本地存储事件结果--TXT
        file_path = os.path.join(self.folder_path, "event_output.txt")
        f = open(file_path, 'w')

        now_list = {}
        # 存储拥堵事件信息
        congestion_list = {}
        # 循环监听心跳
        last_time = time.time()
        while True:
            try:
                time.sleep(0.01)
                # 本地当前时间
                now_time = datetime.now()
                # 心跳状态上报
                if time.time() - last_time >= 1:
                    m = {
                        "timestamp": str(int(time.time())),
                        "status": 0,
                        "type": 1
                    }
                    self.kafka_producer.send(self.config_data["kafka_status"], value=m, key="key")
                    last_time = time.time()

                # 分发门架数据到各个区间队列中
                while not self.q.empty():
                    message = self.q.get(timeout=0.1)
                    if self.config_data["congestion_flag"]:
                        # 上传数据到拥堵算法中
                        self.congestion_kako_queue.put(message, block=False)
                    gantry_id = message[self.define["GANTRY"]]
                    # 过滤掉车牌为空的数据
                    if message[self.define["PLATE"]][:2] in ["车牌", "默A"]:
                        continue
                    if gantry_id in self.id_dict:
                        info_list = self.id_dict[gantry_id]
                        for info in info_list:
                            self.q_dict[info].put(message, block=False)

                # 上报事故信息到kafka
                msg = []
                msg_log = ""
                for info in self.gantry_info.keys():
                    gantry_up = self.gantry_info[info]['up_gantry_id']
                    gantry_down = self.gantry_info[info]['down_gantry_id']
                    up_status = all(x == 0 for x in self.gantry_status[gantry_up])
                    down_status = all(x == 0 for x in self.gantry_status[gantry_down])
                    while not self.block_dict[info].empty():
                        con = self.block_dict[info].get(timeout=0.1)
                        print(info, str(con), datetime.now())
                        tt = con['time'].strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
                        if con['content'] != "畅通":
                            event_id = self.generate_id(info, "block", con['time'], con['content'])
                            s = "阻断"
                            # 阻断上报
                            msg_block = {
                                "orgCode": gantry_down,
                                "upOrgCode": gantry_up,
                                "intervalSpeed": con['speed'],
                                "totalFlow": con['num'],
                                "congestion": s.encode('unicode-escape').decode(),
                                "globalTime": tt,
                                "lastTimeIndex": 1,
                                "info": "",
                                "updateTime": tt,
                                "timestamp": self.active_events[int(info)]["block"]["start_time"],
                                "direction": 0,
                                "sNum": self.kako_dict[gantry_up]['RealPileNumber'],
                                "eNum": self.kako_dict[gantry_down]['RealPileNumber'],
                                "sLon": self.kako_dict[gantry_up]['lon'],
                                "sLat": self.kako_dict[gantry_up]['lat'],
                                "eLon": self.kako_dict[gantry_down]['lon'],
                                "eLat": self.kako_dict[gantry_down]['lat'],
                                "secLevel": 4 if con['con'] == "畅通" else 5,
                                "isBreakdown": 1,
                                "eventId": event_id,
                                "eventInfo": con['content']
                            }
                            if up_status and down_status:
                                msg.append(msg_block)
                        if con['con'] != "畅通" and not self.config_data["congestion_flag"]:
                            event_id = self.generate_id(info, "congestion", con['time'], con['con'])
                            s = "拥堵"
                            # 拥堵上报
                            msg_congestion = {
                                "orgCode": gantry_down,
                                "upOrgCode": gantry_up,
                                "intervalSpeed": con['speed'],
                                "totalFlow": con['num'],
                                "congestion": s.encode('unicode-escape').decode(),
                                "globalTime": tt,
                                "lastTimeIndex": 1,
                                "info": "",
                                "updateTime": tt,
                                "timestamp": self.active_events[int(info)]["congestion"]["start_time"],
                                "direction": 0,
                                "sNum": self.kako_dict[gantry_up]['RealPileNumber'],
                                "eNum": self.kako_dict[gantry_down]['RealPileNumber'],
                                "sLon": self.kako_dict[gantry_up]['lon'],
                                "sLat": self.kako_dict[gantry_up]['lat'],
                                "eLon": self.kako_dict[gantry_down]['lon'],
                                "eLat": self.kako_dict[gantry_down]['lat'],
                                "secLevel": level_dict[con['con']],
                                "isBreakdown": 0,
                                "eventId": event_id,
                                "eventInfo": con['con']
                            }
                            if up_status and down_status:
                                msg.append(msg_congestion)
                        msg_log = msg_log + f"info: {info}, status: {up_status},{down_status}, data: {str(con)}" + "\n"
                if msg:
                    self.kafka_producer.send(self.config_data["kafka_event_topic"], value=msg, key="key")
                    # 保存事故上报信息到本地txt文件
                    f.write(msg_log)

                # 上报拥堵信息到kafka
                while not self.congestion_queue.empty():
                    congestion = self.congestion_queue.get(timeout=0.1)
                    self.q_congestion.put(congestion, block=False)
                    self.kafka_producer.send(self.config_data["kafka_event_topic"], value=congestion, key="key")
                    # print(str(congestion))
                    # 保存事故上报信息到本地txt文件
                    f.write(str(congestion) + "\n")

                # 将多门架监测数据整合
                for info in self.gantry_info.keys():
                    while not self.kafka_data_dict[info].empty():
                        now_list[info] = self.kafka_data_dict[info].get(timeout=0.1)

                # 将拥堵数据整合
                while not self.q_congestion.empty():
                    congestion_info = self.q_congestion.get(timeout=0.1)
                    for congestion in congestion_info:
                        congestion_sn = congestion['orgCode']
                        if congestion_sn not in self.id_dict.keys():
                            continue
                        congestion_list[sn_info_dict[congestion_sn]] = congestion["congestion"]

                # 上报监测数据到kafka
                if self.gantry_info:
                    first_key = list(self.gantry_info.keys())[0]
                    if (now_time - self.start_time) >= timedelta(minutes=float(self.gantry_info[first_key]['time_move'])):
                        send_data = []
                        for info in self.gantry_info.keys():
                            if info in now_list.keys():
                                gantry_up = self.gantry_info[info]['up_gantry_id']
                                gantry_down = self.gantry_info[info]['down_gantry_id']
                                up_status = all(x == 0 for x in self.gantry_status[gantry_up])
                                down_status = all(x == 0 for x in self.gantry_status[gantry_down])
                                if info in congestion_list.keys():
                                    now_list[info]["congestionFlag"] = congestion_list[info]
                                    if congestion_list[info] != "畅通" and now_list[info]["eventFlag"] == "畅通":
                                        eventLevel = level_dict[congestion_list[info]]
                                    elif congestion_list[info] != "畅通" and now_list[info]["eventFlag"] != "畅通":
                                        eventLevel = level_dict["拥堵+阻断"]
                                    else:
                                        eventLevel = now_list[info]["eventLevel"]
                                    now_list[info]["eventLevel"] = eventLevel
                                    congestion_list[info] = "畅通"
                                if not (up_status and down_status):
                                    now_list[info]["abnormalFlag"] = "异常"
                                send_data.append(now_list[info])
                        send_kafka_data = {
                            "timeStamp": now_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3],
                            "snList": send_data
                        }
                        self.kafka_producer.send(self.config_data["kafka_event_index_topic"], value=send_kafka_data, key="key")
                        now_list = {}
                        congestion_list = {}
                        self.start_time = now_time
                        for gantry in self.id_dict.keys():
                            self.gantry_status[gantry] = [1, 1, 1, 1]

            except BaseException as e:
                print("异常", e)
                self.thread_flag = True
                for info in self.gantry_info.keys():
                    self.flag_dict[info].put(self.thread_flag, block=True)
                f.write(str(e) + "\n")
                f.close()



